package cn.doitedu.rtdw.dash_board;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.TableConfig;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class Job5_订单日清日结大屏看板开发 {

    public static void main(String[] args) {
        // 创建编程入口
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(5000);
        env.getCheckpointConfig().setCheckpointStorage("file:/d:/ckpt");
        env.setParallelism(1);

        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);


        /**
         * 开启提前触发器，配合 滚动窗口，实现cumulate窗口的效果
         */
        TableConfig config = tenv.getConfig();
        config.getConfiguration().setBoolean("table.exec.emit.early-fire.enabled", true);
        config.getConfiguration().setString("table.exec.emit.early-fire.delay", "5000 ms");


        // 建cdc连接器表，映射mysql中的 订单表
        // 1. 建表映射 业务库中的  oms_order表
        tenv.executeSql(
                "CREATE TABLE order_mysql (    " +
                        // 订单id
                        " id            INT ,             " +
                        // 订单状态  0：待支付  1:待发货  2: 已发货  3：已完成
                        " status        INT ,             " +
                        // 订单金额
                        " total_amount  decimal(10,0) ,   " +
                        // 订单应付、实付额
                        " pay_amount    decimal(10,0) ,   " +
                        // 订单创建时间
                        " create_time   timestamp(3)  ,   " +
                        // 订单支付时间
                        " payment_time  timestamp(3)  ,   " +
                        // 订单发货时间
                        " delivery_time timestamp(3)  ,   " +
                        // 订单确认时间
                        " receive_time  timestamp(3)  ,   " +
                        // 数据更新时间
                        " modify_time   timestamp(3)  ,   " +
                        " rt as create_time ,             " +
                        " watermark for rt as rt - interval '0' second ,   " +
                        " PRIMARY KEY (id) NOT ENFORCED            " +
                        " ) WITH (                                 " +
                        "     'connector' = 'mysql-cdc',               " +
                        "     'hostname' = 'doitedu'   ,               " +
                        "     'port' = '3306'          ,               " +
                        "     'username' = 'root'      ,               " +
                        "     'password' = 'root'      ,               " +
                        "     'database-name' = 'realtimedw' ,         " +
                        "     'table-name' = 'oms_order'               " +
                        ")"
        );


        tenv.executeSql(
                " SELECT                                                                                                                   " +
                        "     max(rt) as window_end,                                                                                                " +
                        "     count(1) FILTER(where date_format(create_time,'yyyy-MM-dd')=current_date)  as `订单总数`,                              " +
                        "     sum(total_amount) FILTER(where date_format(create_time,'yyyy-MM-dd')=current_date)  as `订单总额`,                     " +
                        "     sum(pay_amount) FILTER(where date_format(create_time,'yyyy-MM-dd')=current_date)  as `订单应付总额`,                   " +
                        " 	  count(1) FILTER(where date_format(create_time,'yyyy-MM-dd')=current_date and status=0)  as `待支付订单总数`,             " +
                        "     sum(total_amount) FILTER(where date_format(create_time,'yyyy-MM-dd')=current_date and status=0)  as `待支付订单总额`,  " +
                        "                                                                                                                            " +
                        " 	  count(1) FILTER(where date_format(payment_time,'yyyy-MM-dd')=current_date )  as `当日支付订单总数`,                      " +
                        "     sum(total_amount) FILTER(where date_format(payment_time,'yyyy-MM-dd')=current_date)  as `当日支付订单总额`,            " +
                        "                                                                                                                            " +
                        " 	  count(1) FILTER(where date_format(delivery_time,'yyyy-MM-dd')=current_date )  as `当日发货订单总数`,                     " +
                        "     sum(total_amount) FILTER(where date_format(delivery_time,'yyyy-MM-dd')=current_date)  as `当日发货订单总额`,           " +
                        "                                                                                                                            " +
                        " 	  count(1) FILTER(where date_format(receive_time,'yyyy-MM-dd')=current_date )  as `当日确认订单总数`,                      " +
                        "     sum(total_amount) FILTER(where date_format(receive_time,'yyyy-MM-dd')=current_date)  as `当日确认订单总额`             " +
                        " FROM order_mysql                                                                                                           " +
                        " GROUP BY                                                                                                                   " +
                        "     TUMBLE(rt,interval '24' hour)                                                                                          "

        ).print();

    }


}
